package cn.com.idmy.mq.listener;

import cn.com.idmy.mq.coder.MqMsgDecoder;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.lang.reflect.Method;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class ConcurrentlyMsgListener extends AbstractMsgListener implements MessageListenerConcurrently {
    public ConcurrentlyMsgListener(Map<String, Map<String, Class<?>>> messageTypeMap, Map<String, Map<String, Method>> methodMap, Map<String, Map<String, Object>> beanMap, Map<String, Map<String, MqMsgDecoder>> decoderMap) {
        super(messageTypeMap, methodMap, beanMap, decoderMap);
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        Iterator<MessageExt> iterator = messages.iterator();
        if (iterator.hasNext()) {
            MessageExt msg = iterator.next();
            if (invoke(msg)) {
                do {
                    if (iterator.hasNext()) {
                        msg = iterator.next();
                    } else {
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                } while (invoke(msg));
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        } else {
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }
}
